package com.wang.tool.task;

import com.alibaba.fastjson.JSON;
import com.wang.tool.common.SimpleException;
import com.wang.tool.lock.core.LockBody;
import com.wang.tool.lock.LockServer;
import com.wang.tool.monitor.InvokeTemplate;
import com.wang.tool.monitor.LinkLogUtil;
import com.wang.tool.monitor.core.MonitorInfo;
import com.wang.tool.monitor.template.AbsMonitorInvoker;
import com.wang.tool.util.CommonUtil;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.function.Function;

/**
 * @Description 分步任务管理器
 * 处理长耗时任务
 * 基本逻辑
 * 1.先create添加任务，添加任务时会进行任务提交预处理
 * 2.有两种方式完结任务
 * 2.1被动回调(update)，这种主要接受外部回调/消息回调进行更新结果
 * 2.2主动自检(check)，部分情况三方没有任务结束主动回调，只能以自动查询的方式处理，称之自检
 * 3.可选,任务中可挂载关联的任务列表,如果任务完成(status in (2,-2)),则可以触发对应任务的自检而不用等到目标任务自行自检
 * @Author shaopeng
 * @Date 2023/6/2
 * @Version 1.0
 * @see TaskManage#create
 * @see TaskManage#check
 * @see TaskManage#update
 * @see TaskManage#pause
 * @see TaskManage#resume
 * <p>
 * 过程1，2中均会有相关api暴露出来，定义在TaskHandler
 * @see TaskHandler
 * 每次更新task会有钩子函数暴露
 * @see TaskManage#onProcess
 * 可在相关操作后调用api实现关联方自检
 * @see TaskManage#fireRelates
 */
public abstract class TaskManage {


    /**
     * 根据id查询
     * select * from task where id=#{id} and env=#{env}
     *
     * @param taskId 任务id
     * @return
     */
    public abstract Task selectById(Long taskId);

    /**
     * 事务执行
     * 暴露public可以实现外界在需要批量处理task情况下保证事务
     * TransactionTemplate.executeWithoutResult((u) -> executor.execute());
     *
     * @param action 执行任务 非空
     * @return
     */
    public abstract void doWithTrans(Runnable action);


    /*
        一般sql

        1.扫表查询需要自检的任务sql
        <select id="selectNeedCheck" resultMap="resultMap">
        select
        *
        from
        task
        <![CDATA[
        where env = #{env} and status not in(2,-2,-1) and now() > check_time
        order by id desc
          ]]>
    </select>

        2.扫表查询需要触发别的任务的sql
    <select id="selectNeedFire" resultMap="resultMap">
        select
        *
        from
        task
        <![CDATA[
        where env = #{env} and status in(2,-2) and fire = 0 and relate_ids is not null
        order by id desc
          ]]>
    </select>
    * */

    /**
     * 创建任务
     *
     * @param taskBase 非空
     * @param other    为什么有这个参数？init时方便使用，在事务中安全运行
     * @return 非空
     */
    public Task create(TaskBase taskBase, Object other) {
        return InvokeTemplate.monitor("task", "create", new AbsMonitorInvoker<Task>(taskBase, other) {

            private Long taskId;

            @Override
            public void weaveOnBefore(MonitorInfo<Task> monitorInfo) {
                monitorInfo.addDigest("ownerId", taskBase == null ? null : taskBase.getOwnerId());
                monitorInfo.addDigest("handler", taskBase == null ? null : taskBase.getHandler());
                monitorInfo.addDigest("relateIds", taskBase == null ? null : CommonUtil.join(taskBase.getRelateIds(), ","));
            }

            @Override
            public Task internalCall() {
                if (taskBase == null) {
                    throw new SimpleException(SimpleException.PARAM_INVALID, "taskBase can not be null");
                }
                String handlerKey = taskBase.getHandler();
                TaskHandler handler = findHandler(handlerKey);
                if (handler == null) {
                    throw new SimpleException(SimpleException.NOT_FOUND, "not found taskHandler:" + handlerKey);
                }
                Task newTask = new Task();
                newTask.setRelateIds(taskBase.getRelateIds());
                newTask.setTaskName(taskBase.getTaskName());
                newTask.setHandler(handlerKey);
                newTask.setGmtCreate(new Date());
                newTask.setGmtModified(new Date());
                newTask.setOwnerId(taskBase.getOwnerId());
                newTask.setHandleTimes(0);
                newTask.setFire(false);
                newTask.setStatus(TaskStatus.WAIT);
                taskId = insert(newTask);
                if (taskId == null) {
                    throw new SimpleException(SimpleException.UN_EXPECT, "taskId can not be null");
                }

//                加db锁
                if (!tryHandler(taskId)) {
                    LinkLogUtil.newTagLog("TaskTryHandlerFail")
                            .v1(String.valueOf(taskId))
                            .log();
                    throw new SimpleException(SimpleException.LOCK_FAIL, "TaskIsRunningOrComplete");
                }
                LinkLogUtil.newTagLog("TaskTryHandlerSuccess")
                        .v1(String.valueOf(taskId))
                        .log();

//                重新查
                Task task = selectById(taskId);
                if (task == null) {
                    LinkLogUtil.newTagLog("NoSuchTask")
                            .v1(String.valueOf(taskId))
                            .log();
                    return null;
                }
                if (task.judgeIsComplete()) {
                    LinkLogUtil.info("TaskAlreadyComplete, taskId={0}", taskId);
                    return task;
                }

//                业务锁
                LockBody lockBody = handler.generateBizLockBody(task);
                if (lockBody == null) {
                    LinkLogUtil.info("TaskHandlerNoBizLock, taskId={0}", taskId);
                    doWithTrans(() -> {
                        TaskStage stage = handler.pre(task, other);
                        if (stage == null) {
                            LinkLogUtil.newTagLog("TaskStageIsNull")
                                    .v1(String.valueOf(taskId))
                                    .log();
                            return;
                        }
                        if (!doUpdate("create", task, stage)) {
                            throw new SimpleException(SimpleException.SET_ERROR, "update Task error, id:{0}, stage:{1}", task.getId(), stage);
                        }
                    });
                    return task;
                }
                LinkLogUtil.info("TaskHandlerBizLock, taskId={0}, body={1}", taskId, lockBody);
                lockDo(lockBody, () -> {
                    LinkLogUtil.info("TaskHandlerBizLockSuccess, taskId={0}", taskId);
                    doWithTrans(() -> {
                        TaskStage stage = handler.pre(task, other);
                        if (stage == null) {
                            LinkLogUtil.newTagLog("TaskStageIsNull")
                                    .v1(String.valueOf(taskId))
                                    .log();
                            return;
                        }
                        if (!doUpdate("create", task, stage)) {
                            throw new SimpleException(SimpleException.SET_ERROR, "update Task error, id:{0}, stage:{1}", task.getId(), stage);
                        }
                    });
                });
                return task;
            }

            @Override
            protected void internalAfter(MonitorInfo<Task> monitorInfo) throws Throwable {
                monitorInfo.addDigest("taskId", taskId);
                Task t = monitorInfo.getResult();
                monitorInfo.addDigest("outId", t == null ? null : t.getOutId());
                monitorInfo.addDigest("status", t == null ? null : t.getStatus());
                monitorInfo.addDigest("bizStatus", t == null ? null : t.getBizStatus());

            }

            @Override
            protected boolean isPrintResult() {
                return true;
            }
        });
    }


    /**
     * check 自检
     *
     * @param taskId 非空
     * @return 没查到task 返回null
     */
    public Task check(Long taskId) {
        return InvokeTemplate.monitor("task", "check", new AbsMonitorInvoker<Task>() {
            @Override
            public void weaveOnBefore(MonitorInfo<Task> monitorInfo) {
                monitorInfo.addDigest("taskId", taskId);
            }

            @Override
            public Task internalCall() {
                if (taskId == null) {
                    throw new SimpleException(SimpleException.PARAM_INVALID, "taskId can not be null");
                }
                if (!tryHandler(taskId)) {
                    LinkLogUtil.newTagLog("TaskTryHandlerFail")
                            .v1(String.valueOf(taskId))
                            .log();
                    throw new SimpleException(SimpleException.LOCK_FAIL, "TaskIsRunningOrComplete");
                }
                LinkLogUtil.newTagLog("TaskTryHandlerSuccess")
                        .v1(String.valueOf(taskId))
                        .log();
//                重新查
                Task task = selectById(taskId);
                if (task == null) {
                    LinkLogUtil.newTagLog("NoSuchTask")
                            .v1(String.valueOf(taskId))
                            .log();
                    return null;
                }
                if (task.judgeIsComplete()) {
                    LinkLogUtil.info("TaskAlreadyComplete, taskId={0}", taskId);
                    return task;
                }


                String handlerKey = task.getHandler();
                TaskHandler handler = findHandler(handlerKey);
                if (handler == null) {
                    throw new SimpleException(SimpleException.NOT_FOUND, "not found taskHandler:" + handlerKey);
                }

//                业务锁
                LockBody lockBody = handler.generateBizLockBody(task);
                if (lockBody == null) {
                    LinkLogUtil.info("TaskHandlerNoBizLock, taskId={0}", taskId);
                    doWithTrans(() -> {
                        TaskStage stage = handler.check(task);
                        if (stage == null) {
                            LinkLogUtil.newTagLog("TaskStageIsNull")
                                    .v1(String.valueOf(taskId))
                                    .log();
                            return;
                        }
                        if (!doUpdate("check", task, stage)) {
                            throw new SimpleException(SimpleException.SET_ERROR, "update Task error, id:{0}, stage:{1}", task.getId(), stage);
                        }
                    });
                    return task;
                }
                LinkLogUtil.info("TaskHandlerBizLock, taskId={0}, body={1}", taskId, lockBody);
                lockDo(lockBody, () -> {
                    LinkLogUtil.info("TaskHandlerBizLockSuccess, taskId={0}", taskId);
                    doWithTrans(() -> {
                        TaskStage stage = handler.check(task);
                        if (stage == null) {
                            LinkLogUtil.newTagLog("TaskStageIsNull")
                                    .v1(String.valueOf(taskId))
                                    .log();
                            return;
                        }
                        if (!doUpdate("check", task, stage)) {
                            throw new SimpleException(SimpleException.SET_ERROR, "update Task error, id:{0}, stage:{1}", task.getId(), stage);
                        }
                    });
                });
                return task;
            }

            @Override
            protected void internalAfter(MonitorInfo<Task> monitorInfo) {
                Task t = monitorInfo.getResult();
                monitorInfo.addDigest("outId", t == null ? null : t.getOutId());
                monitorInfo.addDigest("status", t == null ? null : t.getStatus());
                monitorInfo.addDigest("bizStatus", t == null ? null : t.getBizStatus());
            }
        });
    }


    /**
     * 更新任务状态
     *
     * @param taskId   非空
     * @param function 该函数在事务中执行
     * @return
     */
    public Task update(Long taskId, Function<Task, TaskStage> function) {
        return InvokeTemplate.monitor("task", "update", new AbsMonitorInvoker<Task>() {
            @Override
            public void weaveOnBefore(MonitorInfo<Task> monitorInfo) throws Throwable {
                monitorInfo.addDigest("taskId", taskId);
            }

            @Override
            public Task internalCall() {
                if (taskId == null) {
                    throw new SimpleException(SimpleException.PARAM_INVALID, "taskId can not be null");
                }
                if (!tryHandler(taskId)) {
                    LinkLogUtil.newTagLog("TaskTryHandlerFail")
                            .v1(String.valueOf(taskId))
                            .log();
                    throw new SimpleException(SimpleException.LOCK_FAIL, "TaskIsRunningOrComplete");
                }
                LinkLogUtil.newTagLog("TaskTryHandlerSuccess")
                        .v1(String.valueOf(taskId))
                        .log();
//                重新查
                Task task = selectById(taskId);
                if (task == null) {
                    LinkLogUtil.newTagLog("NoSuchTask")
                            .v1(String.valueOf(taskId))
                            .log();
                    throw new SimpleException(SimpleException.NOT_FOUND, "task:{0} is not exist", taskId);
                }
                if (task.judgeIsComplete()) {
                    LinkLogUtil.info("TaskAlreadyComplete, taskId={0}", taskId);
                    throw new SimpleException(SimpleException.STATUS_INVALID, "task:{0} is already complete", taskId);
                }


                String handlerKey = task.getHandler();
                TaskHandler handler = findHandler(handlerKey);
                if (handler == null) {
                    throw new SimpleException(SimpleException.NOT_FOUND, "not found taskHandler:" + handlerKey);
                }

//                业务锁
                LockBody lockBody = handler.generateBizLockBody(task);
                if (lockBody == null) {
                    LinkLogUtil.info("TaskHandlerNoBizLock, taskId={0}", taskId);
                    doWithTrans(() -> {
                        TaskStage stage = function.apply(task);
                        if (stage == null) {
                            LinkLogUtil.newTagLog("TaskStageIsNull")
                                    .v1(String.valueOf(taskId))
                                    .log();
                            return;
                        }
                        if (!doUpdate("update", task, stage)) {
                            throw new SimpleException(SimpleException.SET_ERROR, "update Task error, id:{0}, stage:{1}", task.getId(), stage);
                        }
                    });
                    return task;
                }
                LinkLogUtil.info("TaskHandlerBizLock, taskId={0}, body={1}", taskId, lockBody);
                lockDo(lockBody, () -> {
                    LinkLogUtil.info("TaskHandlerBizLockSuccess, taskId={0}", taskId);
                    doWithTrans(() -> {
                        TaskStage stage = function.apply(task);
                        if (stage == null) {
                            LinkLogUtil.newTagLog("TaskStageIsNull")
                                    .v1(String.valueOf(taskId))
                                    .log();
                            return;
                        }
                        if (!doUpdate("update", task, stage)) {
                            throw new SimpleException(SimpleException.SET_ERROR, "update Task error, id:{0}, stage:{1}", task.getId(), stage);
                        }
                    });
                });
                return task;
            }

            @Override
            protected void internalAfter(MonitorInfo<Task> monitorInfo) {
                Task t = monitorInfo.getResult();
                monitorInfo.addDigest("outId", t == null ? null : t.getOutId());
                monitorInfo.addDigest("status", t == null ? null : t.getStatus());
                monitorInfo.addDigest("bizStatus", t == null ? null : t.getBizStatus());
            }
        });
    }


    /**
     * 暂停任务
     *
     * @param taskId 非空
     * @return 任务
     */
    public Task pause(Long taskId) {
        return InvokeTemplate.monitor("task", "pause", new AbsMonitorInvoker<Task>() {

            @Override
            public void weaveOnBefore(MonitorInfo<Task> monitorInfo) {
                monitorInfo.addDigest("taskId", taskId);
            }

            @Override
            public Task internalCall() {
                Task task = selectById(taskId);
                if (task == null) {
                    throw new SimpleException(SimpleException.NOT_FOUND, "task:{0} is not exist", taskId);
                }
                if (task.judgeIsComplete()) {
                    return task;
                }
                doWithTrans(() -> doUpdate("pause", task, task.toStage().pause().bizStatus("暂停")));
                return task;
            }

            @Override
            protected void internalAfter(MonitorInfo<Task> monitorInfo) {
                monitorInfo.addDigest("pause", monitorInfo.getResult() == null ? null : monitorInfo.getResult().getStatus() == TaskStatus.PAUSE);
            }
        });
    }


    /**
     * 恢复暂停的任务
     *
     * @param taskId 非空
     * @param checkTime 下次验证时间 非空
     * @return 任务
     */
    public Task resume(Long taskId, Date checkTime) {
        return InvokeTemplate.monitor("task", "resume", new AbsMonitorInvoker<Task>() {

            @Override
            public void weaveOnBefore(MonitorInfo<Task> monitorInfo) {
                monitorInfo.addDigest("taskId", taskId);
            }

            @Override
            public Task internalCall() {
                if (taskId == null || checkTime == null) {
                    throw new SimpleException(SimpleException.PARAM_INVALID, "taskId and checkTime can not be null");

                }
                Task task = selectById(taskId);
                if (task == null) {
                    throw new SimpleException(SimpleException.NOT_FOUND, "task:{0} is not exist", taskId);
                }
                if (task.judgeIsComplete()) {
                    return task;
                }
                doWithTrans(() -> doUpdate("resume", task, task.toStage().nextTime(checkTime)));
                return task;
            }

            @Override
            protected void internalAfter(MonitorInfo<Task> monitorInfo) {
                monitorInfo.addDigest("resume", monitorInfo.getResult() == null ? null : monitorInfo.getResult().getStatus() == TaskStatus.WAIT);
            }
        });
    }


    /**
     * 触发关联方
     * <p>
     * 如果被触发任务也完成了,那么会继续触发关联方
     *
     * @param task 非空 && status in (2,-2)
     * @return 触发有哪些关联任务
     */
    public List<Long> fireRelates(Task task) {
        return InvokeTemplate.monitor("task", "fireRelates", new AbsMonitorInvoker<List<Long>>() {
            @Override
            public void weaveOnBefore(MonitorInfo<List<Long>> monitorInfo) {
                monitorInfo.addDigest("taskId", task == null ? null : task.getId());
            }

            @Override
            protected boolean isPrintResult() {
                return true;
            }

            @Override
            public List<Long> internalCall() {
                if (task == null || !task.judgeIsComplete() || task.getId() == null) {
                    throw new SimpleException(SimpleException.PARAM_INVALID, "task:{} is null or is not complete", task == null ? null : task.getId());
                }
                if (task.isFire()) {
                    LinkLogUtil.newTagLog("TaskAlreadyFire")
                            .v1(String.valueOf(task))
                            .log();
                    return new ArrayList<>(2);
                }
                List<Long> relateIds = task.getRelateIds();
                if (CommonUtil.isEmpty(relateIds)) {
                    return new ArrayList<>(2);
                }
                for (Long relateId : relateIds) {
                    try {
                        Task check = check(relateId);
                        if (check != null && task.judgeIsComplete()) {
                            fireRelates(check);
                        }
                    } catch (Throwable e) {
                        LinkLogUtil.newTagLog("TaskFireRelateError")
                                .v1(String.valueOf(task.getId()))
                                .v2(String.valueOf(relateId))
                                .addAttachment(MonitorInfo.ERROR, e)
                                .log();
                    }
                }
//                修改db为已触发
                if (!updateFireOnComplete(task.getId())) {
//                    非核心不抛异常
                    LinkLogUtil.newTagLog("TaskFireSetError")
                            .v1(String.valueOf(task.getId()))
                            .log();
                } else {
                    task.setFire(true);
                }
                return relateIds;
            }
        });
    }


    /**
     * 插入
     * 要求：返回值非空
     * <p>
     * insert task (...) values(...)
     *
     * @param task 非空
     * @return taskId  非空
     */
    protected abstract Long insert(Task task);


    /**
     * UPDATE
     * task
     * <set>
     * status = 1,
     * gmt_modified = now()
     * </set>
     * <![CDATA[
     * where id=#{taskId}
     * and (status = 0 or (status = 1 and gmt_modified < DATE_SUB(NOW(), INTERVAL 5 MINUTE)))
     * ]]>
     *
     * @param taskId 非空
     * @return
     */
    protected abstract boolean tryHandler(Long taskId);


    /**
     * 更新状态
     * 更新前：status not in (2,-2)
     * 更新后stage内容全覆盖(不需要判断非空)
     * <p>
     * UPDATE
     * task
     * <set>
     * status = #{stage.status},
     * biz_status = #{stage.bizStatus},
     * check_time = #{stage.checkTime},
     * out_id = #{stage.outId},
     * model = #{stage.model,typeHandler=com.alibaba.alsc.merchant.agi.framework.mybatis.JsonTypeHandler},
     * gmt_modified = now(),
     * handle_times = handle_times + 1
     * </set>
     * <![CDATA[
     * where id = #{stage.id} and env = #{env} and handle_times= #{currentHandleTimes} and status not in (2,-2)
     * ]]>
     *
     * @param stage              可能为空
     * @param currentHandleTimes 当前尝试次数
     * @return
     */
    protected abstract boolean updateStageOnNotComplete(TaskStage stage, int currentHandleTimes);

    /**
     * 更新已完成任务将其标记为已fire
     * 更新前：status in (2,-2)
     * 更新后：fire = true
     * <p>
     * UPDATE
     * task
     * <set>
     * fire = 1,
     * gmt_modified = now()
     * </set>
     * <![CDATA[
     * where id = #{id} and env = #{env} and status in (2,-2)
     * ]]>
     *
     * @param taskId 非空
     * @return
     */
    protected abstract boolean updateFireOnComplete(Long taskId);


    /**
     * 获取处理器
     * 可Spring bean 注入管理
     *
     * @param handler 不保证非空，为了提高容错率，对于空key可返回空实现
     * @return
     */
    protected abstract TaskHandler<?> findHandler(String handler);

    /**
     * 加锁执行
     * 建议依赖锁服务
     * 可对接锁服务
     *
     * @param body   非空
     * @param action 非空
     * @return
     * @see LockServer#lockDo
     */
    protected abstract void lockDo(LockBody body, Runnable action);


    /**
     * 任务被处理时，这时db已经更新成功了但是还在事务中
     * 例如version>100  就直接废掉
     * complete后发消息
     *
     * @param method 执行的函数 create / update  / check / pause / recover
     * @param task   非空 && complete in (2,-2)
     * @throws Throwable 内部异常会被吞
     */
    protected abstract void onProcess(String method, Task task) throws Throwable;


    /**
     * 更新状态
     * 整个过程应该被包裹在事务中，外界控制
     *
     * @param task  非空
     * @param stage 非空
     * @return 是否成功
     */
    protected boolean doUpdate(String method, Task task, TaskStage stage) {
        stage.setId(task.getId());
        if (stage.getStatus() == null || stage.getStatus() == TaskStatus.RUNNING) {
            throw new IllegalArgumentException("请设置任务状态");
        }

        if (!updateStageOnNotComplete(stage, task.getHandleTimes())) {
            return false;
        }
        task.setStatus(stage.getStatus());
        task.setCheckTime(stage.getCheckTime());
        task.setOutId(stage.getOutId());
        task.setModel(stage.getModel() == null ? null : JSON.parseObject(JSON.toJSONString(stage.getModel())));
        task.setBizStatus(stage.getBizStatus());
        task.setGmtModified(new Date());
        task.setHandleTimes(task.getHandleTimes() + 1);
        try {
            onProcess(method, task);
        } catch (Throwable e) {
            LinkLogUtil.newTagLog("TaskOnProcessError")
                    .v1(String.valueOf(task.getId()))
                    .addAttachment(MonitorInfo.ERROR, e)
                    .log();
        }
        return true;
    }


}
